package com.gandong8.trace.tool.consumer.rocketmq;


import com.gandong8.trace.tool.util.JsonUtil;
import com.gandong8.trace.tool.util.TraceUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public abstract class AbstractRocketMqListener<T> implements MessageListenerConcurrently  {
    private Class<T> recordClazz;

    public AbstractRocketMqListener() {
        parseClazz();
    }

    /**
     * 默认msg里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
     * 不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS
     * @param msgList
     * @param consumeConcurrentlyContext
     * @return
     */
    @SuppressWarnings("unchecked")
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        if (CollectionUtils.isEmpty(msgList)) {
            log.error("empty message!");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        for(MessageExt messageExt:msgList) {
            TraceUtil.setLogTrace(messageExt.getUserProperty(TraceUtil.TRACE_TAG));
            String topic = messageExt.getTopic();
            String tags = messageExt.getTags();
            String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
            log.info("收到消息topic={}, tags={}, 消息内容={}", topic, tags, body);
            try {
                if ("String".equals(recordClazz.getSimpleName())) {
                    execute((T) body);
                } else {
                    T obj = JsonUtil.toObj(body, recordClazz);
                    execute(obj);
                }
            } catch (Exception e) {
                log.error("消息处理异常", e);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }


    protected abstract void execute(T message);



    @SuppressWarnings("unchecked")
    private void parseClazz() {
        Type type = getClass().getGenericSuperclass();
        if (type instanceof ParameterizedType) {
            recordClazz = (Class<T>) ((ParameterizedType) type).getActualTypeArguments()[0];
        }
    }



}
